package com.xiaoze.testapi;

import com.xiaoze.entity.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author xiaoze
 * @date 2021/12/22
 */
public class TableApiExample {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 1.读取数据
        String path = System.getProperty("user.dir") + "\\src\\main\\resources\\sensor.txt";
        DataStreamSource<String> dataStream = env.readTextFile(path);

        // 2.转换成POJO
        SingleOutputStreamOperator<SensorReading> inputStream = dataStream.map(line -> {
            String[] strArr = line.split(",");
            return new SensorReading(strArr[0], new Long(strArr[1]), new Double(strArr[2]));
        });

        // 3.创建表的环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 4.基于流创建一张表
        Table dataTable = tableEnv.fromDataStream(inputStream);

        // 5.调用table API进行转换操作
        Table result = dataTable
                .select($("id"), $("timeNum"), $("temperature"))
                .where($("id").isEqual("s10"));

        // 6.打印
        result.execute().print();

    }

}
